package com.example.flinkcourse.lesson1.transform;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 自定义 RichMapFunction 转换函数
 * 实现富函数功能，包括状态管理和累加器
 */
public class MyRichFunction extends RichMapFunction<String, String> {
    private static final Logger LOG = LoggerFactory.getLogger(MyRichFunction.class);
    
    // 定义累加器
    private LongCounter processedRecords;
    
    // 定义状态
    private ValueState<Integer> state;

    @Override
    public void open(Configuration parameters) throws Exception {
        LOG.info("Initializing RichFunction...");
        
        // 初始化累加器
        processedRecords = new LongCounter();
        getRuntimeContext().addAccumulator("processed-records", processedRecords);
        
        // 初始化状态
        ValueStateDescriptor<Integer> descriptor = 
            new ValueStateDescriptor<>("processed-count", Integer.class);
        state = getRuntimeContext().getState(descriptor);
        
        LOG.info("RichFunction initialized successfully");
    }

    @Override
    public String map(String value) throws Exception {
        LOG.debug("Processing record: {}", value);
        
        // 更新累加器
        processedRecords.add(1);
        
        // 更新状态
        Integer currentCount = state.value();
        if (currentCount == null) {
            currentCount = 0;
        }
        state.update(currentCount + 1);
        
        // 示例：在值前添加处理计数
        String result = String.format("[%d] %s", currentCount + 1, value);
        
        LOG.debug("Transformed record: {}", result);
        return result;
    }

    @Override
    public void close() throws Exception {
        LOG.info("Closing RichFunction...");
        LOG.info("Total processed records: {}", processedRecords.getLocalValue());
    }
} 